package data.dblp;

import java.io.BufferedReader;
import java.io.ByteArrayInputStream;
import java.io.IOException;
import java.io.InputStreamReader;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Map.Entry;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileStatus;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;

import regular.Initial;

public class PartitionMapper extends Mapper<Text, Text, NullWritable, Text> {

	public void map(Text key, Text value, Context context) throws IOException,
			InterruptedException {
		
		HashSet<Integer> listV = getListIds(value);

		HashMap<Integer, Node> listNodes = getAllNodes(Initial.PATH_ALL_NODE,
				context.getConfiguration());

		// System.out.println(listNodes.keySet().size());

		ArrayList<String> listOutNodes = new ArrayList<String>();
		HashSet<Integer> listCheckOut = new HashSet<Integer>();
		StringBuilder output = new StringBuilder();
		String node = "";
		for (Integer v : listV) {
			node = "";
			// check v is in-node?
			boolean check = false;
			for (Entry<Integer, Node> item : listNodes.entrySet()) {
				int k = item.getKey();
				Node n = item.getValue();
				if (!listV.contains(k)) {
					if (n.neighbors.contains(v))
						check = true;
				}
				if (check)
					break;
			}
			// end check v in in-node

			// set in - out node
			Node vi = listNodes.get(v);

			if (vi != null) {
				node = vi.id + ":";
				if (check) {
					node += "1\t";
				} else
					node += "0\t";

				node += vi.label;

				for (int s : vi.neighbors) {
					if (!listCheckOut.contains(s)) {
						if (!listV.contains(s)) {
							Node a = listNodes.get(s);
							if (a != null) {
								listOutNodes.add(s + ":2\t" + a.label);
							} else
								listOutNodes.add(s + ":2");
						}
						listCheckOut.add(s);

					}
					node += "\t" + s;
				}
			} else {
				if (check) {
					node = v + ":1";
				} else
					node = v + ":0";
			}

			output.append(node + "\n");
		}

		for (String outnode : listOutNodes)
			output.append(outnode + "\n");

		context.write(NullWritable.get(), new Text(output.toString()));

	}

	public HashSet<Integer> getListIds(Text value) throws IOException {
		HashSet<Integer> listIds = new HashSet<Integer>();

		byte[] bytes = value.getBytes();
		ByteArrayInputStream arr = new ByteArrayInputStream(bytes);
		BufferedReader bNode = new BufferedReader(new InputStreamReader(arr));

		String strLine = null;
		while ((strLine = bNode.readLine()) != null) {
			listIds.add(Integer.parseInt(strLine));
		}
		bNode.close();
		return listIds;
	}

	public HashMap<Integer, Node> getAllNodes(String pathDir, Configuration conf)
			throws IOException {

		HashMap<Integer, Node> listNodes = new HashMap<Integer, Node>();
		// HashSet<Integer> listIds = new HashSet<Integer>();

		FileSystem fs = FileSystem.get(conf);
		Path hdfsPath = new Path(pathDir);
		FileStatus[] status = fs.listStatus(hdfsPath);

		for (int f = 0; f < status.length; f++) {
			BufferedReader bNode = new BufferedReader(new InputStreamReader(
					fs.open(status[f].getPath())));
			String strLine = null;

			while ((strLine = bNode.readLine()) != null) {
				String[] cells = strLine.split("\t");
				ArrayList<Integer> neighbors = new ArrayList<Integer>();
				String attributes = "";
				for (int i = 1; i < 6; i++)
					attributes += cells[i] + "\t";

				if (attributes.length() > 0)
					attributes = attributes.substring(0,
							attributes.length() - 1);

				for (int i = 6; i < cells.length; i++)
					neighbors.add(Integer.parseInt(cells[i]));
				
				listNodes.put(Integer.parseInt(cells[0]), new Node(Integer.parseInt(cells[0]), attributes,
						neighbors));				
			}
			bNode.close();
		}
		return listNodes;
	}

}
